package com.example.prop;

import com.example.exp.MySubscriberAsyncExceptionHandler;
import com.example.exp.MySubscriberExceptionHandler;
import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.EventBus;
import org.springframework.stereotype.Service;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * eventBus 消息总线
 */
@Service
public class EventDispatcherImpl implements EventDispatcher {

    /**
     * 同步事件总线  同步执行  谁先注册到EventBus的，谁先执行 若同时订阅则收到事件的顺序跟方法名有关
     */
    private EventBus eventBus = new EventBus(new MySubscriberExceptionHandler());

    /**
     * 异步事件总线  不会等待事件消费方是否收到，直接执行自己后面的代码
     */
    private AsyncEventBus asyncEventBus = new AsyncEventBus(executor(), new MySubscriberAsyncExceptionHandler());

    @Override
    public void register(Object object) {
        eventBus.register(object);
    }

    @Override
    public void unregister(Object object) {
        eventBus.unregister(object);
    }

    @Override
    public void post(Object event) {
        eventBus.post(event);
    }

    @Override
    public void registerAsync(Object object) {
        asyncEventBus.register(object);
    }

    @Override
    public void unregisterAsync(Object object) {
        asyncEventBus.unregister(object);
    }

    @Override
    public void postAsync(Object event) {
        // 事务统一提交
//        if (TransactionSynchronizationManager.isActualTransactionActive()) {
//            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
//                @Override
//                public void afterCommit() {
//                    super.afterCommit();
//                    asyncEventBus.post(event);
//                }
//            });
//        } else {
        asyncEventBus.post(event);
//        }
    }

    /**
     * 指定任务线程池
     *
     * @return
     */
    private ThreadPoolExecutor executor() {
        return new ThreadPoolExecutor(20, 20,
                60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(), r -> new Thread(r, "Async_EventPool_" + r.hashCode()));
    }
}
